package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_etl.pojos.PageAccessTimeBean;
import cn.doitedu.rtdw.data_etl.udfs.TimeTruncUdf;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

public class Job3_访问时长olap分析主题轻度聚合模型 {
    public static void main(String[] args) throws Exception {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 读数据，因为要用api编程，所以用source来读数据得到dataStream
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd_events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("iloveu")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        DataStreamSource<String> dwdEvents = env.fromSource(source, WatermarkStrategy.noWatermarks(), "dwd_events");

        // json解析，提取必要字段，并封装为自定义的javaBean
        SingleOutputStreamOperator<PageAccessTimeBean> beanStream = dwdEvents.map(new RichMapFunction<String, PageAccessTimeBean>() {
            PageAccessTimeBean accessTimeBean;

            @Override
            public void open(Configuration parameters) throws Exception {
                accessTimeBean = new PageAccessTimeBean();
            }

            @Override
            public PageAccessTimeBean map(String dataJson) throws Exception {
                JSONObject jsonObject = JSON.parseObject(dataJson);
                String event_id = jsonObject.getString("event_id");
                String page_type = jsonObject.getString("page_type");
                String device_type = jsonObject.getString("device_type");
                String release_channel = jsonObject.getString("release_channel");
                Long user_id = jsonObject.getLong("user_id");
                String session_id = jsonObject.getString("session_id");

                String page_url = jsonObject.getJSONObject("properties").getString("url");
                Long event_time = jsonObject.getLong("event_time");

                accessTimeBean.of(event_id, page_type, device_type, release_channel, user_id, session_id, page_url, event_time, null);

                return accessTimeBean;
            }
        });


        // 核心逻辑
        SingleOutputStreamOperator<PageAccessTimeBean> processed = beanStream.keyBy(PageAccessTimeBean::getSession_id)
                .process(new KeyedProcessFunction<String, PageAccessTimeBean, PageAccessTimeBean>() {

                    ValueState<PageAccessTimeBean> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<PageAccessTimeBean> desc = new ValueStateDescriptor<>("s", PageAccessTimeBean.class);
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.minutes(30))
                                .updateTtlOnCreateAndWrite()
                                .build();
                        desc.enableTimeToLive(ttlConfig);


                        state = getRuntimeContext().getState(desc);
                    }

                    @Override
                    public void processElement(PageAccessTimeBean bean, KeyedProcessFunction<String, PageAccessTimeBean, PageAccessTimeBean>.Context ctx, Collector<PageAccessTimeBean> out) throws Exception {

                        PageAccessTimeBean stateBean = state.value();

                        // 如果状态为空，则将进来的这第一条数据放入状态
                        if (stateBean == null) {
                            // 填充页面打开时间，然后放入状态
                            bean.setPage_open_time(bean.getEvent_time());
                            state.update(bean);
                        }
                        // 如果状态不为空，且进来的事件上pageLoad，则要生成虚拟事件
                        else if (bean.getEvent_id().equals("page_load")) {
                            // 将当前最新eventTime替换掉状态里面的eventTime，但别的字段（所在页面、所在页面的打开时间）不变
                            state.value().setEvent_time(bean.getEvent_time());

                            // 并输出一次
                            out.collect(state.value());

                            // 然后将本次数据填充页面打开时间，并放入状态，作为一个新的页面生命周期的开始
                            bean.setPage_open_time(bean.getEvent_time());
                            state.update(bean);

                        } else if (bean.getEvent_id().equals("wake_up")) {
                            // 更新状态数据中的页面打开时间
                            state.value().setPage_open_time(bean.getEvent_time());
                            // 更新状态数据中的eventTime
                            state.value().setEvent_time(bean.getEvent_time());

                        } else if(bean.getEvent_id().equals("app_close")){
                            state.value().setEvent_time(bean.getEvent_time());

                            // 清理状态中的数据
                            state.clear();


                        }
                        else {
                            // 其他的事件，不需要特别处理，只要将状态中的eventTime更新成最新的自己的时间即可
                            state.value().setEvent_time(bean.getEvent_time());

                        }

                        // 并输出状态中的数据
                        out.collect(state.value());
                    }
                });


        // 将虚拟事件插值法加工好的数据，转成表
        tenv.createTemporaryView("tmp",processed,
                Schema.newBuilder()
                        .column("page_type", DataTypes.STRING())
                        .column("device_type",DataTypes.STRING())
                        .column("release_channel",DataTypes.STRING())
                        .column("user_id",DataTypes.BIGINT())
                        .column("session_id",DataTypes.STRING())
                        .column("page_url",DataTypes.STRING())
                        .column("event_time",DataTypes.BIGINT())
                        .column("page_open_time",DataTypes.BIGINT())
                        .columnByExpression("rt","to_timestamp_ltz(event_time,3)")
                        .watermark("rt","rt - interval '0' second")
                .build());


        tenv.executeSql("select * from tmp");
/*
+-----------+-------------+-----------------+---------+------------+------------+---------------+--------------------+-------------------------+
| page_type | device_type | release_channel | user_id | session_id |   page_url |    event_time |     page_open_time |                      rt |
+-----------+-------------+-----------------+---------+------------+------------+---------------+--------------------+-------------------------+
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |      index |   1695803951000 |   1695803951000 | 2023-09-27 16:39:11.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |      index |   1695803952000 |   1695803951000 | 2023-09-27 16:39:12.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |      index |   1695803953000 |   1695803951000 | 2023-09-27 16:39:13.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |      index |   1695803954000 |   1695803951000 | 2023-09-27 16:39:14.000 |

|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page08 |   1695803954000 |   1695803954000 | 2023-09-27 16:39:14.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page08 |   1695803955000 |   1695803954000 | 2023-09-27 16:39:15.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page08 |   1695803956000 |   1695803954000 | 2023-09-27 16:39:16.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page08 |   1695803957000 |   1695803954000 | 2023-09-27 16:39:17.000 |

|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803957000 |   1695803957000 | 2023-09-27 16:39:17.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803958000 |   1695803957000 | 2023-09-27 16:39:18.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803959000 |   1695803957000 | 2023-09-27 16:39:19.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803960000 |   1695803957000 | 2023-09-27 16:39:20.000 |

|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803980000 |   1695803980000 | 2023-09-27 16:39:40.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803981000 |   1695803980000 | 2023-09-27 16:39:41.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803982000 |   1695803980000 | 2023-09-27 16:39:42.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s10 |     page09 |   1695803983000 |   1695803980000 | 2023-09-27 16:39:43.000 |
|    文章页  |         mi8 |    华为应用市场 |          3 |        s11 |     page08 |   1695804983000 |   1695804983000 | 2023-09-27 16:56:23.000 |

 */






        /**
         * 计算 每一个页面，每一次访问的时长
         */
        tenv.executeSql(
                " CREATE TEMPORARY VIEW tmp3 AS                               "+
                        " with tmp2 as (                                              "+
                        "     SELECT                                                  "+
                        "       *                                                     "+
                        "     from tmp                                                "+
                        "     where page_url is not null                              "+
                        " )                                                           "+

                        " SELECT                                                      "+
                        "     window_start,                                           "+
                        " 	window_end,                                               "+
                        " 	page_type,                                                "+
                        " 	device_type,                                              "+
                        " 	release_channel,                                          "+
                        " 	user_id,                                                  "+
                        " 	session_id,                                               "+
                        " 	page_url,                                                 "+
                        " 	page_open_time,                                           "+
                        " 	max(event_time)-min(event_time) as page_access_time       "+
                        " from table(                                                 "+
                        "     tumble(table tmp2,descriptor(rt),interval '1' minute)   "+
                        " )                                                           "+
                        " group by                                                    "+
                        "     window_start,                                           "+
                        " 	window_end,                                               "+
                        " 	page_type,                                                "+
                        " 	device_type,                                              "+
                        " 	release_channel,                                          "+
                        " 	user_id,                                                  "+
                        " 	session_id,                                               "+
                        " 	page_url,                                                 "+
                        " 	page_open_time                                            "
        );

        //tenv.executeSql("select * from tmp3").print();
/*
+----+-------------------------+-------------------------+----------+--------------+-----------------+--------+-----------+-------------+-----------------+------------------+
| op |            window_start |              window_end |page_type |  device_type | release_channel |user_id |session_id |    page_url |  page_open_time | page_access_time |
+----+-------------------------+-------------------------+----------+--------------+-----------------+--------+-----------+-------------+-----------------+------------------+
| +I | 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |           mi8 |       华为应用市场 |    3 |    s10      |        index | 1695803951000 |              3000 |
| +I | 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |           mi8 |       华为应用市场 |    3 |    s10      |       page08 | 1695803954000 |              3000 |
| +I | 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |           mi8 |       华为应用市场 |    3 |    s10      |       page09 | 1695803980000 |              3000 |
| +I | 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |           mi8 |       华为应用市场 |    3 |    s10      |       page09 | 1695803957000 |              3000 |

 */


        /**
         *   按照  用户，页面类型，页面，终端等维度，聚合 “页面访问总时长”，“页面访问总人数”，“页面访问总次数”，“页面访问单次最小时长”，“页面访问单次最大时长”，
         */
        tenv.createTemporaryFunction("time_trunc", TimeTruncUdf.class);
        tenv.executeSql(
                " CREATE TEMPORARY VIEW tmp4  AS                          "+
                        " SELECT                                                   "+
                        " 	to_date(date_format(window_end,'yyyy-MM-dd'))  as dt,  "+
                        " 	page_type,                                             "+
                        " 	page_url,                                              "+
                        " 	device_type,                                           "+
                        " 	release_channel,                                       "+
                        " 	time_trunc(window_end,60) as time_60m,                 "+
                        " 	sum(page_access_time) as amt_access_time,              "+
                        " 	count(1) as amt_access_count,                          "+
                        " 	user_id as amt_access_users,                           "+
                        " 	min(page_access_time) as min_access_time,              "+
                        " 	max(page_access_time) as max_access_time               "+
                        " from tmp3                                                "+
                        " group by                                                 "+
                        "     window_start,                                        "+
                        " 	window_end,                                            "+
                        " 	page_type,                                             "+
                        " 	device_type,                                           "+
                        " 	release_channel,                                       "+
                        " 	user_id,                                               "+
                        " 	page_url                                               "
        );

        //tenv.executeSql("select * from tmp4").print();
/*
+-------------------------+-------------------------+----------+------------+----------------+---------+----------------+---------------+-----------------+----------------+-----------------+
|            window_start |              window_end |page_type |device_type |release_channel |page_url |amt_access_time |amt_access_cnt |amt_access_users |min_access_time | max_access_time |
+-------------------------+-------------------------+----------+------------+----------------+---------+----------------+---------------+-----------------+----------------+-----------------+
| 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |         mi8 |     华为应用市场 |   index |            3000 |             1 |               3 |           3000 |            3000 |
| 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |         mi8 |     华为应用市场 |  page09 |            6000 |             2 |               3 |           3000 |            3000 |
| 2023-09-27 16:39:00.000 | 2023-09-27 16:40:00.000 |   文章页 |         mi8 |     华为应用市场 |  page08 |            3000 |             1 |               3 |           3000 |            3000 |
 */


        /**
         * 创建一个逻辑映射表，映射doris中目标物理表
         * 这里面，要做user_id => bitmap的映射配置
         */
        tenv.executeSql(
                " CREATE TABLE   acc_timelong_page_doris(              "+
                        "     dt                 DATE  ,                       "+
                        "     page_type          STRING,                       "+
                        "     page_url           STRING,                       "+
                        "     device_type        STRING,                       "+
                        "     release_channel    STRING,                       "+
                        "     time_60m           STRING,                       "+
                        "     amt_access_time    BIGINT,                       "+
                        "     amt_access_count   BIGINT,                       "+
                        "     amt_access_users   BIGINT,                       "+
                        "     min_access_time    BIGINT,                       "+
                        "     max_access_time    BIGINT                        "+
                        " ) WITH (                                             "+
                        "    'connector' = 'doris',                            "+
                        "    'fenodes' = 'doitedu:8030',                       "+
                        "    'table.identifier' = 'dws.acc_timelong_page',     "+
                        "    'username' = 'root',                              "+
                        "    'password' = 'root',                              "+
                        "    'sink.label-prefix' = 'doris_label-oooo',         "+
                        "    'sink.properties.columns' = 'dt,page_type,page_url,device_type,release_channel,time_60m,amt_access_time,amt_access_count,amt_access_users,min_access_time,max_access_time,amt_access_users=to_bitmap(amt_access_users)'   "+
                        " )                                                    "
        );


        tenv.executeSql("insert into acc_timelong_page_doris select * from tmp4");


        env.execute();

    }

}
